

package com.hazelcast.jet.impl.observer;

import com.hazelcast.jet.function.Observer;

import javax.annotation.Nonnull;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import static com.hazelcast.jet.impl.util.ExceptionUtil.rethrow;

public final class BlockingIteratorObserver<T> implements Iterator<T>, Observer<T> {
    private static final Object COMPLETED = new Object();

    private final BlockingQueue<Object> itemQueue;
    private Object next;

    public BlockingIteratorObserver() {
        this.itemQueue = new LinkedBlockingQueue<>();
    }

    @Override
    public void onNext(@Nonnull T t) {
        itemQueue.add(t);
    }

    @Override
    public void onError(@Nonnull Throwable throwable) {
        itemQueue.add(WrappedThrowable.of(throwable));
    }

    @Override
    public void onComplete() {
        itemQueue.add(COMPLETED);
    }


    @Override
    public boolean hasNext() {
        if (next == null) {
            next = waitForNext();
        }

        if (next instanceof WrappedThrowable wrappedThrowable) {
            throw rethrow(wrappedThrowable.get());
        } else {
            return next != COMPLETED;
        }
    }

    @Nonnull
    private Object waitForNext() {
        try {
            return itemQueue.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw rethrow(e);
        }
    }

    @Override
    @Nonnull
    public T next() {
        if (!hasNext()) {
            throw new NoSuchElementException();
        }
        T item = (T) next;
        next = null;
        return item;
    }
}
